# coding=utf-8
from __future__ import absolute_import  # 这个必须放入在首行
import re
import time
import pymysql
import subprocess
from datetime import date, timedelta

from celery import Celery
from celery.task.schedules import crontab

from celery_tasks.batchTask import BatchCenter
# 某个程序中出现的队列，在broker中不存在，则立刻创建它
CELERY_CREATE_MISSING_QUEUES = True
CELERY_IMPORTS = ("async_task.tasks", "async_task.notify")
# 创建celery客户端
app = Celery('tasks', broker='redis://127.0.0.1:6379/0')
CELERYD_CONCURRENCY = 20  # 并发worker数
CELERYD_FORCE_EXECV = True  # 非常重要,有些情况下可以防止死锁
CELERYD_PREFETCH_MULTIPLIER = 1
CELERYD_MAX_TASKS_PER_CHILD = 100  # 每个worker最多执行万100个任务就会被销毁，可防止内存泄露
# CELERYD_TASK_TIME_LIMIT = 60    # 单个任务的运行时间不超过此值，否则会被SIGKILL 信号杀死
# BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 90}
# 任务发出后，经过一段时间还未收到acknowledge , 就将任务重新交给其他worker执行
# CELERY_DISABLE_RATE_LIMITS = True
app.conf.update(
    #  配置所在时区
    CELERY_BROKER_URL='redis://localhost:6379/1',
    CELERY_RESULT_BACKEND='redis://localhost:6379/0',
    CELERY_TIMEZONE='Asia/Shanghai',
    CELERY_ENABLE_UTC=True,
    #  官网推荐消息序列化方式为json
    CELERY_ACCEPT_CONTENT=['json'],
    CELERY_TASK_SERIALIZER='json',
    CELERY_RESULT_SERIALIZER='json',
    # CELERY_IGNORE_RESULT=True,
    # CELERY_SEND_EVENTS=False,
    # 配置定时任务
    CELERYBEAT_SCHEDULE={
        'daydisk': {
            'task': 'celery_tasks.tasks.daydisk',  # tasks.py模块下的函数方法
            'schedule': crontab(minute=0, hour=1),
        },
        'daypartition': {
            'task': 'celery_tasks.tasks.createPartition',  # tasks.py模块下的函数方法
            'schedule': crontab(minute=0, hour=0),
        },
    }
)


@app.task
def daydisk():
    """硬盘空间不足告警"""
    disks = timeout_command('df | grep storage')
    # disks = '100.1G     30.5G    3.2M    1%  /storage'
    pt = re.compile(r' (\d+)')
    lefts = pt.findall(disks)
    if lefts:
        num = float(lefts[2]) / 1024
        if num < 100:
            db1 = pymysql.connect(host="127.0.0.1", user="insec", passwd="InSec888!", db="safeplatform", charset='utf8')
            cursor1 = db1.cursor()
            cursor1.execute(
                "SELECT partition_name from information_schema.partitions where table_name='t_device_log' limit 1")
            res = cursor1.fetchone()
            try:
                sql = "alter table t_device_log drop partition %s" % res[0]
                cursor1.execute(sql)
                db1.commit()
            except Exception as e:
                db1.rollback()
            cursor1.close()
            db1.close()


daydisk.delay()

@app.task
def createPartition():
    """定时创建今天及明天分区"""
    day = date.today()
    curday = day.strftime('%Y%m%d')
    tomorrow = day + timedelta(days=1)
    tomorrowstr = tomorrow.strftime('%Y%m%d')
    after = day + timedelta(days=2)
    afterstr = after.strftime('%Y%m%d')

    db = pymysql.connect(host='127.0.0.1', user='insec', password='InSec888!', database='safeplatform', charset='utf8')
    cursor = db.cursor()
    checksql1 = "select partition_name from information_schema.partitions where table_name='t_device_log' order by partition_name desc limit 1"
    cursor.execute(checksql1)
    res = cursor.fetchone()
    dbday = res[0]
    if 'p%s' % curday != dbday and 'p%s' % tomorrowstr != dbday:
        sql = "alter table t_device_log add partition(partition p%s values less than(to_days('%s')))" % (curday, tomorrowstr)
        cursor.execute(sql)
        db.commit()
        # print('today partition is buliding')

    if 'p%s' % tomorrowstr != dbday:
        sql1 = "alter table t_device_log add partition(partition p%s values less than(to_days('%s')))" % (tomorrowstr, afterstr)
        cursor.execute(sql1)
        db.commit()
        # print('tomorrow partition is buliding')
    cursor.close()

    db.close()

createPartition.delay()

def timeout_command(command, timeout=3):
    start = int(time.time())
    process = subprocess.Popen(command, bufsize=10000, shell=True, stdout=subprocess.PIPE, close_fds=True)
    while process.poll() is None:
        time.sleep(0.1)
        now = int(time.time())
        if now - start > timeout:
            try:
                process.terminate()
            except Exception as e:
                return ''
            return ''
    out = process.communicate()[0]
    if process.stdin:
        process.stdin.close()
    if process.stdout:
        process.stdout.close()
    if process.stderr:
        process.stderr.close()
    try:
        process.kill()
    except OSError:
        pass
    if out:
        return bytes.decode(out)
    else:
        return ''


@app.task
def trigger_batch(_type, _ids, _bhr, _batch_id):
    """触发任务调用"""
    batch = BatchCenter(_type, _ids, _bhr, _batch_id)
    batch.center()
    del batch
